/**
 * Created by lzfto
 * Created time: 2025/05/25
 */
package org.lzfto.flink.demo.business.flinkTest.flinkObjects.connector.jdbc;

import java.util.List;
import java.util.Optional;

import javax.annotation.Nullable;

import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.RowLevelModificationScanContext;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete;
import org.lzfto.flink.demo.business.flinkTest.base.record.JdbcOptionRecord;

public class CustomJdbcSink implements DynamicTableSink, SupportsRowLevelDelete {
    private JdbcOptionRecord  jdbcOption;

    public CustomJdbcSink(JdbcOptionRecord  option) {
        this.jdbcOption = option;
    }

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(DynamicTableSink.Context context) {
        return new CustomJdbcSinkRuntimeProvider(this.jdbcOption);
    }

    @Override
    public DynamicTableSink copy() {
        CustomJdbcSink copy = new CustomJdbcSink(this.jdbcOption);
        return copy;
    }

    @Override
    public String asSummaryString() {
        return "Custom JDBC Connector";
    }

    @Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
    	//支持各种模式？
        return ChangelogMode.all();
    }

    @Override
    public RowLevelDeleteInfo applyRowLevelDelete(@Nullable RowLevelModificationScanContext context) {
        return new RowLevelDeleteInfo() {
            @Override
            public Optional<List<Column>> requiredColumns() {
                return Optional.empty();
            }

            @Override
            public RowLevelDeleteMode getRowLevelDeleteMode() {
                return RowLevelDeleteMode.DELETED_ROWS;
            }
        };
    }
}
